package com.bml.architect.base;

import com.bml.architect.utils.MqConstants;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;

/**
 * 同步发送，发送者需要得到接受者返回的状态，从而进行判断消息是否发送成功
 */
public class SyncProducer {

    public static void main(String[] args) throws MQClientException, UnsupportedEncodingException, RemotingException, InterruptedException, MQBrokerException {

        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
        producer.setNamesrvAddr(MqConstants.getNameServerAddr());
        producer.start();
        System.out.println("rocketmq连接成功...");
        //发送消息
        for (int i = 0; i < 10; i++) {
            /**
             * 第一个参数：主题，每一个消息都有一个主题，这个主题可以是事先创建好的主题，也可以自动创建，但是需要配置
             * 参数
             * 第二个参数：消息的TAG，Tag的作用可以让消费者监听感兴趣的tag，也就是用于消息过滤使用
             * 第三个参数：消息的key，通过key进行归类
             * 第四个参数：消息的内容主体
             */

            Message message = new Message("TopicTest", "TagA", "simplekey11", ("hello rocketmq ,synchroized send  " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            //同步传递消息，消息会发给集群中的一个Broker节点,broker节点接受到过后会返回一个发送对象，包含了发送状态
            SendResult send = producer.send(message);
            if (send.getSendStatus().equals(SendStatus.SEND_OK)) {
                System.out.println("消息发送成功");
            } else {
                System.out.println("消息发送失败");
            }
        }
        System.out.println("消息发送完毕");
        producer.shutdown();

    }
}
